package test.mq;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.TimeUnit;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

public class testrocket2_Producer {
	public static void main(String[] args) throws Exception {
		AsyncProducer();
	}
	

	    public static void AsyncProducer() throws Exception {
	        //Instantiate with a producer group name.
	        DefaultMQProducer producer = new
	            DefaultMQProducer("group1");
	        // Specify name server addresses.
	        producer.setNamesrvAddr("192.168.0.147:9876");
	        //Launch the instance.
	    	producer.setVipChannelEnabled(false);
		
			// 发送失败以后重试次数
			producer.setRetryTimesWhenSendFailed(3);
			// 发送别名
			producer.setUnitName("p_1" );
	        producer.start();
	        for (int i = 0; i < 5; i++) {
	            //Create a message instance, specifying topic, tag and message body.
	            Message msg = new Message("test" /* Topic */,
	                "TagA" /* Tag */,
	                ("Hello RocketMQ " +
	                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
	            );
	            //Call send message to deliver message to one of brokers.
	            SendResult sendResult = producer.send(msg);
	            System.out.printf("%s%n", sendResult);
	        }
	        //Shut down once the producer instance is not longer in use.
	        producer.shutdown();
	    
	    }
	
	   public static void test3() throws Exception{
	        //Instantiate with a producer group name.
	        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
	        // Specify name server addresses.
	        producer.setNamesrvAddr("192.168.0.147:9876");
	        //Launch the instance.
	        producer.start();
	        for (int i = 0; i < 100; i++) {
	            //Create a message instance, specifying topic, tag and message body.
	            Message msg = new Message("TopicTest" /* Topic */,
	                "TagA" /* Tag */,
	                ("Hello RocketMQ " +
	                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
	            );
	            //Call send message to deliver message to one of brokers.
	            producer.sendOneway(msg);
	        }
	        //Wait for sending to complete
	        Thread.sleep(5000);        
	        producer.shutdown();
	    }
	public static void test() throws MQClientException, UnsupportedEncodingException, RemotingException,
			MQBrokerException, InterruptedException {
		// Instantiate with a producer group name.
		DefaultMQProducer producer = new DefaultMQProducer("group2");
		// producer.setCreateTopicKey(createTopicKey);
		// Specify name server addresses.
		producer.setNamesrvAddr("192.168.0.147:9876");
		// Launch the instance.
		producer.start();
		for (int i = 0; i < 100; i++) {
			// Create a message instance, specifying topic, tag and message body.
			Message msg = new Message("TopicTest" /* Topic */, "TagA" /* Tag */,
					("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
			);
			// Call send message to deliver message to one of brokers.
			SendResult sendResult = producer.send(msg);
			System.out.printf("%s%n", sendResult);
		}
		// Shut down once the producer instance is not longer in use.
		producer.shutdown();

	}

	/**
	 * 
	 * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例
	 * 
	 * 注意：ProducerGroupName需要由应用来保证唯一
	 * 
	 * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
	 * 
	 * 因为服务器会回查这个Group下的任意一个Producer
	 * 
	 */
	public static void test2() throws MQClientException, UnsupportedEncodingException, RemotingException,
			MQBrokerException, InterruptedException {

		final DefaultMQProducer producer = new DefaultMQProducer("group2");

		producer.setNamesrvAddr("192.168.0.147:9876");

		// producer.setInstanceName("Producer");

		/**
		 * 
		 * Producer对象在使用之前必须要调用start初始化，初始化一次即可
		 * 
		 * 注意：切记不可以在每次发送消息时，都调用start方法
		 * 
		 */

		producer.start();

		/**
		 * 
		 * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。
		 * 
		 * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，
		 * 
		 * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，
		 * 
		 * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。
		 * 
		 */

		for (int i = 0; i < 10; i++) {
			try {
				{
					Message msg = new Message("test1", // topic

							"TagA", // tag

							"OrderID001", // key

							("Hello MetaQA").getBytes());// body

					SendResult sendResult = producer.send(msg);

					System.out.println(sendResult);

				}

				{
					Message msg = new Message("test1", // topic

							"TagB", // tag

							"OrderID0034", // key

							("Hello MetaQB").getBytes());// body

					SendResult sendResult = producer.send(msg);

					System.out.println(sendResult);

				}

				{
					Message msg = new Message("test1", // topic

							"TagC", // tag

							"OrderID061", // key

							("Hello MetaQC").getBytes());// body

					SendResult sendResult = producer.send(msg);

					System.out.println(sendResult);

				}

			}

			catch (Exception e) {
				e.printStackTrace();

			}

			TimeUnit.MILLISECONDS.sleep(1000);

		}

		/**
		 * 
		 * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己
		 * 
		 * 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法
		 * 
		 */

		// producer.shutdown();

		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
			public void run() {
				producer.shutdown();

			}

		}));

		System.exit(0);

	}

}
